import time
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Tuple, Optional
from 新版2.指标计算.指标配置 import MACDConfig


##MACD信号计算器类
class MACDSignalCalculator:

    def __init__(self, xtdata, use_warmup: bool = True):
        """
        初始化MACD信号计算器

        Args:
            use_warmup: 是否使用预热数据，默认True
        """
        self.use_warmup = use_warmup
        self.xtdata = xtdata

    ##获取前N天的日线数据用于预热
    def get_historical_data_for_warmup(self, stock_code: str, days: int = 5) -> Optional[pd.DataFrame]:
        try:
            end_date = datetime.now().strftime("%Y%m%d")
            start_date = (datetime.now() - timedelta(days=days)).strftime("%Y%m%d")

            self.xtdata.download_history_data(stock_code=stock_code, period="1d", start_time=start_date,
                                              end_time=end_date)
            time.sleep(1)

            data = self.xtdata.get_local_data(stock_list=[stock_code], period='1d', start_time=start_date,
                                              end_time=end_date)
            if data and stock_code in data:
                return data[stock_code]
            return None
        except Exception:
            return None

    ##生成预热数据，模拟开盘前的价格波动
    def generate_warmup_data(self, latest_close: float, warmup_hours: int = 2) -> pd.DataFrame:
        # 从8:00开始，每5分钟一个数据点
        warmup_points = warmup_hours * 12  # 2小时 * 12个5分钟点

        # 基于最新收盘价生成预热数据
        base_price = latest_close

        # 生成时间序列（从8:00开始）
        start_time = datetime.now().replace(hour=8, minute=0, second=0, microsecond=0)
        warmup_times = [start_time + timedelta(minutes=5 * i) for i in range(warmup_points)]

        # 生成价格数据（模拟小幅波动）
        np.random.seed(42)  # 固定随机种子，确保可重现
        price_changes = np.random.normal(0, base_price * 0.001, warmup_points)  # 0.1%的标准差
        warmup_prices = [base_price + change for change in price_changes]

        # 创建预热数据DataFrame
        warmup_df = pd.DataFrame({
            'time': warmup_times,
            'price': warmup_prices,
            'open': warmup_prices,
            'high': [p * (1 + abs(np.random.normal(0, 0.0005))) for p in warmup_prices],
            'low': [p * (1 - abs(np.random.normal(0, 0.0005))) for p in warmup_prices],
            'close': warmup_prices,
            'volume': [int(np.random.uniform(100, 1000)) for _ in range(warmup_points)]
        })

        return warmup_df

    ##支持预热数据的EMA计算，实现热启动
    def enhanced_ema_with_warmup(self, series: pd.Series, period: int,warmup_data: Optional[pd.DataFrame] = None) -> pd.Series:
        if not isinstance(series, pd.Series):
            series = pd.Series(series)

        if warmup_data is not None and not warmup_data.empty:
            # 使用预热数据初始化EMA
            warmup_series = warmup_data['close'] if 'close' in warmup_data.columns else warmup_data['price']
            if len(warmup_series) >= period:
                # 从预热数据计算初始EMA值
                initial_ema = warmup_series.tail(period).mean()
                # 从预热数据的最后一个值开始计算
                last_warmup_price = warmup_series.iloc[-1]
            else:
                # 预热数据不足，使用简单平均
                initial_ema = warmup_series.mean() if not warmup_series.empty else series.iloc[0]
                last_warmup_price = initial_ema
        else:
            # 无预热数据，使用传统方法
            initial_ema = series.iloc[0]
            last_warmup_price = initial_ema

        # 计算EMA
        alpha = 2 / (period + 1)
        ema_values = [initial_ema]

        for price in series:
            new_ema = alpha * price + (1 - alpha) * ema_values[-1]
            ema_values.append(new_ema)

        # 返回EMA值（去掉初始值）
        return pd.Series(ema_values[1:], index=series.index)

    ##支持预热数据的MACD计算，实现热启动"
    def enhanced_calculate_macd(self, prices: pd.Series, fast: int = 8, slow: int = 17, signal: int = 9,warmup_data: Optional[pd.DataFrame] = None) -> Tuple[pd.Series, pd.Series, pd.Series]:
        if not isinstance(prices, pd.Series):
            prices = pd.Series(prices)

        # 确保为float并使用顺序索引
        series = prices.astype(float).reset_index(drop=True)

        # 使用增强版EMA计算
        ema_short = self.enhanced_ema_with_warmup(series, fast, warmup_data)
        ema_slow = self.enhanced_ema_with_warmup(series, slow, warmup_data)

        # 计算DIFF
        diff = ema_short - ema_slow

        # 使用DIFF计算DEA（信号线）
        if warmup_data is not None and not warmup_data.empty:
            # 如果有预热数据，使用预热数据计算预热DIFF来初始化DEA
            warmup_series = warmup_data['close'] if 'close' in warmup_data.columns else warmup_data['price']
            if len(warmup_series) >= max(fast, slow):
                # 计算预热数据的EMA
                warmup_ema_short = self.enhanced_ema_with_warmup(warmup_series, fast)
                warmup_ema_slow = self.enhanced_ema_with_warmup(warmup_series, slow)
                # 计算预热数据的DIFF
                warmup_diff_series = warmup_ema_short - warmup_ema_slow
                # 使用预热数据DIFF的最后一个值来初始化DEA
                warmup_diff_last = warmup_diff_series.iloc[-1] if not warmup_diff_series.empty else 0
            else:
                warmup_diff_last = 0

            # 使用预热DIFF的最后一个值来初始化DEA计算
            dea = self.enhanced_ema_with_warmup(diff, signal, pd.DataFrame({'close': [warmup_diff_last]}))
        else:
            dea = self.enhanced_ema_with_warmup(diff, signal)

        # 计算MACD柱状图
        macd_histogram = 2 * (diff - dea)

        return diff, dea, macd_histogram

    ##将tick数据重新采样为分钟数据
    def resample_ticks_to_minutes(self, df: pd.DataFrame, price_col: str = 'price') -> pd.DataFrame:
        if 'time' not in df.columns:
            return pd.DataFrame()

        df_resampled = df.set_index('time')
        df_minutes = df_resampled[price_col].resample('1min').agg({
            'open': 'first',
            'high': 'max',
            'low': 'min',
            'close': 'last',
            'volume': 'sum' if 'volume' in df_resampled.columns else lambda x: len(x)
        }).dropna()
        return df_minutes.reset_index()

    ##下载并获取tick数据
    def get_segment_ticks(self, stock_code: str) -> Optional[Dict[str, pd.DataFrame]]:
        today = datetime.now().strftime("%Y%m%d")
        # today = "20250811"
        self.xtdata.download_history_data(stock_code=stock_code, period="tick", start_time=today, end_time=today)
        time.sleep(2)

        try:
            data = self.xtdata.get_local_data(stock_list=[stock_code], period='tick', start_time=today, end_time=today)
            return data if data and stock_code in data else None
        except Exception:
            return None

    ##传统MACD计算函数，支持预热数据
    def calculate_macd(self, prices: pd.Series, fast: int = 8, slow: int = 17, signal: int = 9,warmup_data: Optional[pd.DataFrame] = None) -> Tuple[pd.Series, pd.Series, pd.Series]:
        if warmup_data is not None and not warmup_data.empty:
            # 使用增强版MACD计算
            return self.enhanced_calculate_macd(prices, fast, slow, signal, warmup_data)

        if not isinstance(prices, pd.Series):
            prices = pd.Series(prices)
        # 确保为float并使用顺序索引，便于逐步递推
        series = prices.astype(float).reset_index(drop=True)

        def ths_ema(series_in: pd.Series, period: int) -> pd.Series:
            if len(series_in) < period:
                return pd.Series([np.nan] * len(series_in), index=series_in.index)
            ema = series_in.astype(float).copy()
            # 前period-1为NaN，第period个点为SMA初始化
            ema.iloc[: period - 1] = np.nan
            sma_init = series_in.iloc[:period].mean()
            ema.iloc[period - 1] = sma_init
            alpha = 2 / (period + 1)
            for i in range(period, len(series_in)):
                ema.iloc[i] = alpha * series_in.iloc[i] + (1 - alpha) * ema.iloc[i - 1]
            return ema

        ema_short = ths_ema(series, fast)
        ema_slow = ths_ema(series, slow)
        diff = ema_short - ema_slow
        valid_diff = diff.dropna()
        if len(valid_diff) < signal:
            dea_full = pd.Series([np.nan] * len(series))
            macd_full = pd.Series([np.nan] * len(series))
            return diff, dea_full, macd_full
        dea_valid = ths_ema(valid_diff, signal)
        dea_full = pd.Series([np.nan] * len(series))
        dea_full.iloc[dea_valid.index] = dea_valid.values
        macd_valid = 2 * (valid_diff.loc[dea_valid.index] - dea_valid)
        macd_full = pd.Series([np.nan] * len(series))
        macd_full.iloc[macd_valid.index] = macd_valid.values
        return diff, dea_full, macd_full

    ##检测MACD金叉死叉信号并返回结果
    def detect_macd_signals(self, stock_code: str, fast: int = 12, slow: int = 26, signal: int = 9,use_warmup: Optional[bool] = None) -> Dict[str, List[Dict[str, any]]]:
        """
        检测MACD金叉死叉信号并返回结果

        Args:
            stock_code: 股票代码

        Returns:
            包含金叉和死叉信号的字典，格式如下：
            {
                'golden_cross': [{'time': '2024-01-01 09:30:00', 'price': 10.5}, ...],
                'death_cross': [{'time': '2024-01-01 10:15:00', 'price': 10.3}, ...],
                'latest_macd': {'diff': 0.001, 'dea': -0.002, 'macd': 0.006, 'trend': 'bullish'},
                'data_info': {'total_minutes': 240, 'time_range': '09:30-15:00'}
            }
        """
        # 获取tick数据
        ticks = self.get_segment_ticks(stock_code)
        if not ticks or stock_code not in ticks:
            return {
                'golden_cross': [],
                'death_cross': [],
                'latest_macd': {},
                'data_info': {}
            }

        df = ticks[stock_code]
        if df.empty:
            return {
                'golden_cross': [],
                'death_cross': [],
                'latest_macd': {},
                'data_info': {}
            }

        # 预热数据处理
        warmup_df = None
        use_warmup_effective = self.use_warmup if use_warmup is None else use_warmup
        if use_warmup_effective:
            try:
                historical_data = self.get_historical_data_for_warmup(stock_code, days=5)
                if historical_data is not None and not historical_data.empty:
                    latest_close = historical_data['close'].iloc[-1] if 'close' in historical_data.columns else \
                        historical_data['price'].iloc[-1]
                    warmup_df = self.generate_warmup_data(latest_close, warmup_hours=2)
            except Exception:
                warmup_df = None

        # 确定价格列
        price_col = next((col for col in ['price', 'lastPrice', 'close', 'last', 'current'] if col in df.columns), None)
        if not price_col:
            return {
                'golden_cross': [],
                'death_cross': [],
                'latest_macd': {},
                'data_info': {}
            }

        # 转换时间戳
        time_col = next((col for col in df.columns if 'time' in col.lower()), None)
        if not time_col:
            return {
                'golden_cross': [],
                'death_cross': [],
                'latest_macd': {},
                'data_info': {}
            }

        try:
            if df[time_col].dtype in ['int64', 'float64']:
                df['time'] = pd.to_datetime(df[time_col], unit='ms') + pd.Timedelta(hours=8)
            else:
                df['time'] = pd.to_datetime(df[time_col], format='%Y%m%d%H%M%S', errors='coerce')

            df = df[df['time'].notna()].sort_values('time')
            if df.empty:
                return {
                    'golden_cross': [],
                    'death_cross': [],
                    'latest_macd': {},
                    'data_info': {}
                }
        except Exception:
            return {
                'golden_cross': [],
                'death_cross': [],
                'latest_macd': {},
                'data_info': {}
            }

        # 过滤无效数据和午休时间
        df_filtered = df[df[price_col] > 0].copy()
        lunch_start = pd.Timestamp('11:30:00').time()
        lunch_end = pd.Timestamp('12:59:59').time()
        df_filtered['time_only'] = df_filtered['time'].dt.time
        df_filtered = df_filtered[
            ~((df_filtered['time_only'] >= lunch_start) & (df_filtered['time_only'] <= lunch_end))]

        if df_filtered.empty:
            return {
                'golden_cross': [],
                'death_cross': [],
                'latest_macd': {},
                'data_info': {}
            }

        # 重新采样为分钟数据
        df_minutes = self.resample_ticks_to_minutes(df_filtered, price_col)
        if df_minutes.empty:
            return {
                'golden_cross': [],
                'death_cross': [],
                'latest_macd': {},
                'data_info': {}
            }

        # 准备预热数据用于MACD计算
        warmup_minutes = None
        if warmup_df is not None:
            warmup_minutes = self.resample_ticks_to_minutes(warmup_df, 'close')

        # 计算MACD（参数可配置）
        macd_line, signal_line, histogram = self.calculate_macd(
            df_minutes['close'], fast=fast, slow=slow, signal=signal, warmup_data=warmup_minutes
        )

        # 检测金叉死叉（鲁棒版，加入容差与斜率确认，避免0附近抖动导致反向提示）
        delta = macd_line - signal_line  # DIF - DEA
        eps = getattr(MACDConfig, 'CROSS_EPSILON', 1e-6)
        min_gap = max(0, int(getattr(MACDConfig, 'MIN_CROSS_GAP_BARS', 1)))

        prev_delta = delta.shift(1)
        delta_change = delta - prev_delta

        # 容差符号：> eps 为正，< -eps 为负，其余视为 0（交汇/近似交汇）
        sign_prev = (prev_delta > eps).astype(int) - (prev_delta < -eps).astype(int)
        sign_now = (delta > eps).astype(int) - (delta < -eps).astype(int)

        # 在存在0的边界时，用斜率来确认方向（只在一侧为0时触发）
        cross_up_mask = (
                ((sign_prev < 0) & (sign_now > 0)) |
                ((sign_prev < 0) & (sign_now == 0) & (delta_change > 0)) |
                ((sign_prev == 0) & (sign_now > 0) & (delta_change > 0))
        )

        cross_down_mask = (
                ((sign_prev > 0) & (sign_now < 0)) |
                ((sign_prev > 0) & (sign_now == 0) & (delta_change < 0)) |
                ((sign_prev == 0) & (sign_now < 0) & (delta_change < 0))
        )

        # 去抖动：限制交叉最小间隔（按bar计）
        def thin_cross(mask_series, min_gap_bars):
            idx = mask_series[mask_series.fillna(False)].index.to_list()
            if not idx:
                return mask_series
            kept = []
            last_kept = None
            for i in idx:
                if last_kept is None or (i - last_kept) >= min_gap_bars:
                    kept.append(i)
                    last_kept = i
            new_mask = mask_series.copy()
            new_mask[:] = False
            new_mask.loc[kept] = True
            return new_mask

        if min_gap > 1:
            cross_up_mask = thin_cross(cross_up_mask, min_gap)
            cross_down_mask = thin_cross(cross_down_mask, min_gap)

        # 构建金叉信号数组
        golden_cross = []
        cross_up_times = df_minutes.loc[cross_up_mask.fillna(False), 'time']
        for t in cross_up_times:
            if pd.isna(t):
                continue
            time_str = pd.to_datetime(t).strftime('%Y-%m-%d %H:%M:%S')
            price_at_time = df_minutes.loc[df_minutes['time'] == t, 'close'].iloc[0] if not df_minutes[
                df_minutes['time'] == t].empty else 0
            golden_cross.append({
                'time': time_str,
                'price': float(price_at_time)
            })

        # 构建死叉信号数组
        death_cross = []
        cross_down_times = df_minutes.loc[cross_down_mask.fillna(False), 'time']
        for t in cross_down_times:
            if pd.isna(t):
                continue
            time_str = pd.to_datetime(t).strftime('%Y-%m-%d %H:%M:%S')
            price_at_time = df_minutes.loc[df_minutes['time'] == t, 'close'].iloc[0] if not df_minutes[
                df_minutes['time'] == t].empty else 0
            death_cross.append({
                'time': time_str,
                'price': float(price_at_time)
            })

        # 构建最新MACD信息（只在有有效数值时返回，避免 NaN 输出）
        latest_macd = {}
        try:
            if not macd_line.empty and not signal_line.empty:
                valid_mask = macd_line.notna() & signal_line.notna()
                if valid_mask.any():
                    last_idx = macd_line[valid_mask].index[-1]
                    latest_diff = macd_line.loc[last_idx]
                    latest_dea = signal_line.loc[last_idx]
                    if not histogram.empty and last_idx in histogram.index and not pd.isna(histogram.loc[last_idx]):
                        latest_histogram = histogram.loc[last_idx]
                    else:
                        latest_histogram = 2 * (latest_diff - latest_dea)

                    latest_macd = {
                        'diff': float(latest_diff),
                        'dea': float(latest_dea),
                        'macd': float(latest_histogram),
                        'trend': 'bullish' if latest_diff > latest_dea else 'bearish'
                    }
        except Exception:
            latest_macd = {}

        # 构建数据信息
        data_info = {
            'total_minutes': len(df_minutes),
            'time_range': f"{df_minutes['time'].min().strftime('%H:%M')}-{df_minutes['time'].max().strftime('%H:%M')}"
        }

        return {
            'golden_cross': golden_cross,
            'death_cross': death_cross,
            'latest_macd': latest_macd,
            'data_info': data_info
        }